package belf.migrate.evaluate.helper;

import belf.migrate.api.exception.CatalogException;
import belf.migrate.api.exception.DatabaseNotExistException;
import belf.migrate.api.exception.TableNotExistException;
import belf.migrate.api.job.Job;
import belf.migrate.api.model.QualifiersModel;
import belf.migrate.api.model.TableInfoModel;
import belf.migrate.api.table.schema.AbstractJdbcCatalog;
import belf.migrate.api.table.schema.ColumnMetaData;
import belf.migrate.api.table.schema.TablePath;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.type.DbObjectType;
import belf.migrate.api.type.LevelType;
import belf.migrate.api.type.LogLevelType;
import belf.migrate.api.type.MatchingType;
import belf.migrate.api.util.DatabaseType;
import belf.migrate.api.util.PostLog;
import belf.migrate.core.jdbc.DBFunctionMapping;
import belf.migrate.core.jdbc.DBTypeMapping;
import belf.migrate.core.jdbc.FunctionModel;
import belf.migrate.core.jdbc.TypeModel;
import belf.migrate.evaluate.model.*;
import belf.migrate.evaluate.type.EvaluateColType;
import belf.migrate.evaluate.type.GrammarType;
import belf.migrate.evaluate.model.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;

import java.util.*;
import java.util.stream.Collectors;

import static belf.migrate.api.job.JobCode.EXECUTING;


@Slf4j
public class EvaluateTableTool extends EvaluateBaseTool {

    private AbstractJdbcCatalog catalog;

    public EvaluateTableTool(JobContext jobContext, Job job) {
        super(jobContext, job);
        this.catalog = jobContext.getSourceCatalog();
    }

    @Override
    public EvaluateTableResult run() {

        DatabaseType sourceDatabaseType = DatabaseType.getDatabaseType(job.getTask().getTaskConf().getSourceConf().getCatalogName());
        if (!(sourceDatabaseType != null && sourceDatabaseType.getObjectTypes().contains(DbObjectType.TABLE))) {
            log.info("当前数据库不支持表评估!");
            return null;
        }

        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "开始进行表对象的评估!",
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());
        log.info("开始进行表对象的评估：");

        Map<String, List<EvaluateTableModel>> tableModelMap = new HashMap<>();
        Map<String, EvaluateSinkResult> sinkResultMap = new HashMap<>();
        EvaluateTableResult result = new EvaluateTableResult();
        Map<String, Map<String, EvaluateColMapping>> databaseColMap = new HashMap<>();
        Map<String, Map<String, List<EvaluateColMapping>>> tableColMapping = new HashMap<>();
        String databaseName = job.getTask().getTaskConf().getSourceConf().getDatabaseName();
        String schemaName = job.getTask().getTaskConf().getSourceConf().getSchemaName();
        String taskType = job.getTask().getTaskConf().getTaskType();
        QualifiersModel qualifiersModel = getSourceQualifiersModel();
        Map<String, Set<GrammarType>> grammarTypesMap = new HashMap<>();

        //用以计算兼容性
        Map<String, Map<String, Integer>> columnRiskMap = new HashMap<>();
        Map<String, Map<String, Integer>> funcRiskMap = new HashMap<>();
        try {
            List<String> tableNameList;
            if (CollectionUtils.isEmpty(job.getMissions())) {
                tableNameList = catalog.listTables(databaseName, schemaName);
            } else {
                tableNameList = job.getMissions();
            }
            result.setAllObjectSize(tableNameList.size());
            double allTableSize = 0;
            int allTableRows = 0;
            for (int index=0; index < tableNameList.size(); index++) {
                jobContext.addProgress(DbObjectType.TABLE.getTypeName(), tableNameList.size());
                long startTime = System.currentTimeMillis();
                try {
                    MatchingType matchingType = MatchingType.FULL_MATCH;
                    TablePath tablePath = TablePath.of(databaseName, schemaName, tableNameList.get(index), qualifiersModel);
                    List<ColumnMetaData> allColumn = catalog.getAllColumnsBySql(tablePath);
                    //获取表的ddl语句
                    String tableDDL = catalog.showCreateTable(tablePath);
                    String[] catalogs = taskType.split("-");

                    //针对一对多评估情况
                    for (int i=0;i<job.getTask().getTaskConf().getEvaluateConf().getSinkDbs().size();i++) {
                        List<EvaluateColMapping> colMappingList = new ArrayList<>();
                        String sinkCatalog = job.getTask().getTaskConf().getEvaluateConf().getSinkDbs().get(i);
                        taskType = String.format("%s-%s", catalogs[0], sinkCatalog);
                        EvaluateSinkResult sinkResult = getSinkResultList(sinkResultMap, sinkCatalog);
                        List<EvaluateTableModel> tableModelList = getTableModelListByMap(tableModelMap, sinkCatalog);
                        taskType = catalogs[0] + "-" + sinkCatalog;
                        //获取源库和目标库的字段映射表
                        Map<String, TypeModel> mapping = DBTypeMapping.getInstance().getMapping(
                                taskType);

                        int bigColsSize = 0;
                        int specialColsSize = 0;
                        int compatibleColsSize = 0;
                        int unKnownColsSize = 0;

                        List<EvaluateColumnModel> bigDataCols = new ArrayList<>();
                        List<EvaluateColumnModel> specialDataCols = new ArrayList<>();
                        List<EvaluateColumnModel> unKnownCols = new ArrayList<>();
                        //评估出源库和目标库的字段兼容性
                        for (ColumnMetaData columnMetaData : allColumn) {

                            if (isNotCompatibleCol(mapping, columnMetaData.getColumnTypeName().toUpperCase(Locale.ROOT))) {

                                //统计不兼容的字段
                                addMap(columnRiskMap, sinkCatalog, columnMetaData.getColumnTypeName().toUpperCase(Locale.ROOT));
                                addDatabaseColMapping(databaseColMap, sinkCatalog, columnMetaData.getColumnTypeName().toUpperCase(Locale.ROOT), new EvaluateColMapping(columnMetaData, false));
                                colMappingList.add(new EvaluateColMapping(columnMetaData, true));
                                TypeModel typeModel = mapping.get(columnMetaData.getColumnTypeName().toUpperCase(Locale.ROOT));
                                if (typeModel != null && LevelType.isIncompatible(typeModel.getLevel())) {
                                    //地理信息类型
                                    if (EvaluateColType.GEO.getType().equals(typeModel.getType())) {
                                        addGrammarType(grammarTypesMap, sinkCatalog, GrammarType.GEO_NOT_COMPATIBLE);
                                    }
                                    //json类型
                                    if (EvaluateColType.JSON.getType().equals(typeModel.getType())) {
                                        addGrammarType(grammarTypesMap, sinkCatalog, GrammarType.JSON_NOT_COMPATIBLE);
                                    }
                                }
                            }

                            if (mapping.containsKey(columnMetaData.getColumnTypeName().toUpperCase(Locale.ROOT)) &&
                                    !catalog.bigDataColumns().contains(columnMetaData.getColumnTypeName().toUpperCase(Locale.ROOT)) &&
                                    !catalog.specialColumns().contains(columnMetaData.getColumnTypeName().toUpperCase(Locale.ROOT))) {
                                compatibleColsSize++;
                            } else {
                                EvaluateColumnModel columnModel = new EvaluateColumnModel();
                                columnModel.setColumnName(columnMetaData.getColumnName());
                                columnModel.setColumnType(columnMetaData.getColumnTypeName());
                                columnModel.setColumnDisplaySize(columnMetaData.getColumnDisplaySize());
                                columnModel.setColumnComment(columnMetaData.getColumnComment());
                                matchingType = MatchingType.CONVERT_MATCH;
                                if (catalog.bigDataColumns().contains(columnMetaData.getColumnTypeName().toUpperCase(Locale.ROOT))) {
                                    bigColsSize++;
                                    bigDataCols.add(columnModel);
                                } else if (catalog.specialColumns().contains(columnMetaData.getColumnTypeName().toUpperCase(Locale.ROOT))) {
                                    specialColsSize++;
                                    specialDataCols.add(columnModel);
                                } else {
                                    unKnownColsSize++;
                                    unKnownCols.add(columnModel);
                                }
                            }

                        }
                        if (!CollectionUtils.isEmpty(colMappingList)) {
                            addTableColMapping(tableColMapping, sinkCatalog, tableNameList.get(index), colMappingList);
                        }

                        //获取函数映射关系
                        List<String> functionIncompatible = catalog.findFunctionIncompatible(
                                tableDDL,
                                catalogs[0],
                                sinkCatalog);

                        Map<String, FunctionModel> funcMapping = DBFunctionMapping.getInstance().getMapping(taskType);
                        List<EvaluateFunctionModel> evaluateFunctionModels = new ArrayList<>();

                        //获取不兼容函数
                        for (String funcName : functionIncompatible) {
                            EvaluateFunctionModel evaluateFunctionModel = new EvaluateFunctionModel();
                            evaluateFunctionModel.setName(funcName);
                            String level = funcMapping.get(funcName).getLevel();
                            if (LevelType.WARNING.getName().equals(level) && matchingType.getId() < 1) {
                                matchingType = MatchingType.CONVERT_MATCH;
                            } else if (LevelType.ERROR.getName().equals(level) && matchingType.getId() < 2) {
                                //统计不兼容的函数
                                addMap(funcRiskMap, sinkCatalog, funcName.toUpperCase(Locale.ROOT));

                                matchingType = MatchingType.NO_MATCH;
                            }
                            evaluateFunctionModel.setLevel(level);
                            evaluateFunctionModel.setDesc(funcMapping.get(funcName).getDescription());
                            evaluateFunctionModels.add(evaluateFunctionModel);
                        }

                        TableInfoModel tableInfoModel = catalog.getTableInfoByName(tablePath);
                        //组建返回实体
                        EvaluateTableModel model = new EvaluateTableModel();
                        model.setDatabaseName(databaseName);
                        model.setSchemaName(schemaName);
                        model.setName(tableNameList.get(index));
                        model.setDdlSql(tableDDL);
                        model.setDataSize(tableInfoModel.getSize());
                        allTableSize += tableInfoModel.getSize();
                        allTableRows += (int) tableInfoModel.getRows();
                        model.setRowSize(tableInfoModel.getRows());
                        model.setFuncDiff(evaluateFunctionModels);
                        model.setMatchType(matchingType.getType());

                        model.setCompatibleColsSize(String.valueOf(compatibleColsSize));
                        model.setBigColsSize(String.valueOf(bigColsSize));
                        model.setSpecialColsSize(String.valueOf(specialColsSize));
                        model.setUnKnowColsSize(String.valueOf(unKnownColsSize));
                        model.setAllColsSize(String.valueOf(allColumn.size()));

                        model.setBigDataCols(bigDataCols);
                        model.setSpecialDataCols(specialDataCols);
                        model.setUnKnowCols(unKnownCols);
                        //计算耗时
                        long endTime = System.currentTimeMillis();
                        String timeConsuming = String.valueOf((double) (endTime - startTime) / 1000);
                        model.setTimeConsuming(String.format("%s秒", timeConsuming));
                        tableModelList.add(model);
                        tableModelMap.put(sinkCatalog, tableModelList);
                        switch (matchingType.getId()) {
                            case 0:
                                sinkResult.addCompatibleSize(tableNameList.size());
                                break;
                            case 1:
                                sinkResult.addConvertSize(tableNameList.size());
                                break;
                            case 2:
                                sinkResult.addUnCompatibleSize(tableNameList.size());
                                break;
                            default:
                                sinkResult.addUnKnownSize(tableNameList.size());
                                break;
                        }
                        sinkResultMap.put(sinkCatalog, sinkResult);
                    }

                } catch (CatalogException | TableNotExistException e) {
                    log.error("表评估失败!", e);
                    PostLog.sendLogModel(
                            job.getJobId(),
                            job.getTask().getTaskType().getName(),
                            String.format("表评估失败: %s", e.getMessage()),
                            LogLevelType.ERROR.getType(),
                            EXECUTING,
                            jobContext.getProgress());
                }
                String curLogMsg = String.format("表 %s 评估已完成，目前进度：%s / %s.", tableNameList.get(index) , index + 1, tableNameList.size());
                PostLog.sendLogModel(
                        job.getJobId(),
                        job.getTask().getTaskType().getName(),
                        curLogMsg,
                        LogLevelType.INFO.getType(),
                        EXECUTING,
                        jobContext.getProgress());
                log.info(curLogMsg);
            }
            result.setObjectList(tableModelMap);
            result.setSinkStatisticsMap(sinkResultMap);
            result.setTableColMapping(tableColMapping);
            result.setDatabaseColMapping(getDatabaseColMapping(databaseColMap));
            result.setAllTableSize(allTableSize);
            result.setAllTableRows(allTableRows);

            result.setGrammarTypes(grammarTypesMap);
            result.setColumnRisk(columnRiskMap);
            result.setFunctionRisk(funcRiskMap);
        } catch (CatalogException | DatabaseNotExistException e) {
            log.error("表评估失败", e);
            PostLog.sendLogModel(
                    job.getJobId(),
                    job.getTask().getTaskType().getName(),
                    String.format("表评估失败: %s", e.getMessage()),
                    LogLevelType.ERROR.getType(),
                    EXECUTING,
                    jobContext.getProgress());
        }
        log.info("表评估成功完成!");
        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "表评估成功完成!",
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());

        return result;
    }

    /**
     * 判断是否是兼容的列
     */
    private boolean isNotCompatibleCol(Map<String, TypeModel> mapping, String colType) {
        if (!mapping.containsKey(colType)) {
            return true;
        }
        TypeModel typeModel = mapping.get(colType);
        return StringUtils.isEmpty(typeModel.getTo()) || LevelType.isIncompatible(typeModel.getLevel());
    }

    /**
     * 获取不同目标端的表评估结果
     */
    public List<EvaluateTableModel> getTableModelListByMap(Map<String, List<EvaluateTableModel>> tableModelMap, String key) {
        if (tableModelMap.containsKey(key)) {
            return tableModelMap.get(key);
        } else {
            return new ArrayList<>();
        }
    }

    /**
     * 添加语法类型
     */
    public void addGrammarType(Map<String, Set<GrammarType>> grammarTypesMap, String sinkCatalog, GrammarType grammarType) {
        Set<GrammarType> grammarTypes;
        if (!grammarTypesMap.containsKey(sinkCatalog)) {
            grammarTypes = new HashSet<>();
        } else {
            grammarTypes = grammarTypesMap.get(sinkCatalog);
        }
        grammarTypes.add(grammarType);
        grammarTypesMap.put(sinkCatalog, grammarTypes);
    }

    /**
     * 添加表级字段映射关系
     */
    public void addTableColMapping(
            Map<String, Map<String, List<EvaluateColMapping>>> tableColMapping,
            String sinkCatalog,
            String tableName,
            List<EvaluateColMapping> colMappingList) {
        Map<String, List<EvaluateColMapping>> colMapping;
        if (!tableColMapping.containsKey(sinkCatalog)) {
            colMapping = new HashMap<>();
        } else {
            colMapping = tableColMapping.get(sinkCatalog);
        }
        List<EvaluateColMapping> evaluateColMappings;
        if (!colMapping.containsKey(tableName)) {
            evaluateColMappings = new ArrayList<>();
        } else {
            evaluateColMappings = colMapping.get(tableName);
        }
        evaluateColMappings.addAll(colMappingList);
        colMapping.put(tableName, evaluateColMappings);
        tableColMapping.put(sinkCatalog, colMapping);
    }

    /**
     * 添加库级别字段映射关系
     */
    public void addDatabaseColMapping(
            Map<String, Map<String, EvaluateColMapping>> databaseColMapping,
            String sinkCatalog,
            String colName,
            EvaluateColMapping evaluateColMapping) {
        Map<String, EvaluateColMapping> colMapping;
        if (!databaseColMapping.containsKey(sinkCatalog)) {
            colMapping = new HashMap<>();
        } else {
            colMapping = databaseColMapping.get(sinkCatalog);
        }
        colMapping.put(colName, evaluateColMapping);
        databaseColMapping.put(sinkCatalog, colMapping);
    }

    /**
     * 库级别字段映射关系转换
     */
    public Map<String, List<EvaluateColMapping>> getDatabaseColMapping(Map<String, Map<String, EvaluateColMapping>> databaseColMapping) {
        return databaseColMapping.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> new ArrayList<>(e.getValue().values())));
    }
}
